package com.swhy

import java.net.URI
import java.util.{Date, UUID}

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper


object ChgeIndxDictionary {
  def main(args: Array[String]): Unit = {
    if (args.length != 2) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
//改下方的sql为需要装换成jason的sql
    val allcompany =s"""
        |select
        |indx_name,
        |indx_no,
        |saving_pd,
        |null as indx_desc,
        |null as business_logic,
        |null as indx_busi_caliber,
        |indx_data as indx_val_data,
        |stat,
        |memo,
        |now() as create_time,
        |now() as update_time
        |from
        |dim_data.dim_dao_accsum_dic
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()

      val indx_name = x.getAs[String]("indx_name") //指标名称
      val indx_no = x.getAs[String]("indx_no") //指标编码
      val saving_pd = x.getAs[String]("saving_pd") //更新周期
      val indx_desc = x.getAs[String]("indx_desc") //更新周期
      val business_logic = x.getAs[String]("business_logic") //更新周期
      val indx_busi_caliber = x.getAs[String]("indx_busi_caliber") //更新周期
      val indx_val_data = x.getAs[String]("indx_val_data") //指标值样例
      val stat = x.getAs[Integer]("stat") //状态
      val memo = x.getAs[String]("memo") //备注
      val create_time = x.getAs[Date]("create_time") //创建时间
      val update_time = x.getAs[Date]("update_time") //修改时间

      val _id = new ID()
      val index = new Index()
      val companyInfo = new ChgeIndxDictionary()

      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)

      companyInfo.setIndxName(indx_name)
      companyInfo.setIndxNo(indx_no)
      companyInfo.setSavingPd(saving_pd)
      companyInfo.setIndxValData(indx_val_data)
      companyInfo.setIndxDesc(indx_desc)
      companyInfo.setBusinessLogic(business_logic)
      companyInfo.setIndxBusiCaliber(indx_busi_caliber)
      companyInfo.setStat(stat)
      companyInfo.setMemo(memo)
      companyInfo.setCreateTime(create_time)
      companyInfo.setUpdateTime(update_time)


      val mapper = new ObjectMapper()

      val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(indexJson + "\n" + companyInfoJson) //json结果串
      cominfo
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.saveAsTextFile(outputPath)
    sc.stop()
  }
}
